package com.chukun.flink.dataset.operator;

import org.apache.flink.api.common.functions.GroupReduceFunction;
import org.apache.flink.api.java.DataSet;
import org.apache.flink.api.java.ExecutionEnvironment;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.util.Collector;

import java.util.*;

/**
 * @author chukun
 * @version 1.0.0
 * @description reduceGroup 操作符的基本操作
 * @createTime 2022年05月28日 20:35:00
 */
public class ReduceGroupOperator {

    public static void main(String[] args) throws Exception {

        ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();

        List<Tuple2<String, Integer>> list = new ArrayList<>();
        list.add(new Tuple2<>("张三", 15));
        list.add(new Tuple2<>("张三", 20));
        list.add(new Tuple2<>("张三", 21));
        list.add(new Tuple2<>("李四", 31));
        list.add(new Tuple2<>("李四", 38));
        list.add(new Tuple2<>("李四", 45));
        list.add(new Tuple2<>("王五", 55));
        list.add(new Tuple2<>("王五", 67));

        DataSet<Tuple2<String, Integer>> sourceDataSet = env.fromCollection(list);

        DataSet<Tuple2<Integer, String>> groupReduceDataSet = sourceDataSet.groupBy((key) -> key.f0)
                .reduceGroup(new GroupReduceFunction<Tuple2<String, Integer>, Tuple2<Integer, String>>() {

                    @Override
                    public void reduce(Iterable<Tuple2<String, Integer>> values, Collector<Tuple2<Integer, String>> collector) throws Exception {
                        Set<String> uniqString = new HashSet<>();
                        Integer key = null;
                        Iterator<Tuple2<String, Integer>> iterator = values.iterator();
                        while (iterator.hasNext()) {
                            Tuple2<String, Integer> next = iterator.next();
                            key = next.f1;
                            uniqString.add(next.f0);
                        }
                        for (String uniq : uniqString) {
                            collector.collect(Tuple2.of(key, uniq));
                        }
                    }
                });

        groupReduceDataSet.print("GroupReduceFunction");

        env.execute("ReduceGroupOperator");
    }
}
